package com.atguigu.flink.sqlfunction;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 *
 *  本质上是聚合，输出不是1行1列，可以是N行N列
 *      求所有同一种类型的传感器中水位的前2名
 *
 *      输入:  s1,100,10  输出  ： s1,10
 *      输入:  s1,100,20  输出  ： 1,20
 *                               2,10
 */
public class Demo5_TableAggFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);
        Table table = tableEnvironment.from("t1");

        tableEnvironment.createTemporaryFunction("myUDTAF",new MyUDTAF());
        //sql中没有，只能在TableAPI中使用
        table.groupBy($("id"))
             .flatAggregate(Expressions.call("myUDTAF",$("vc")))
             .select($("id"),$("rank"),$("vcnum"))
             .execute()
             .print();


    }

    /*
        下面几个 TableAggregateFunction 的方法是必须要实现的：

        createAccumulator()
        accumulate()
     */
    @FunctionHint(output = @DataTypeHint("ROW<rank INT, vcnum INT>"))
    public static class MyUDTAF extends TableAggregateFunction<Row,MyAcc>{

        @Override
        public MyAcc createAccumulator() {
            return new MyAcc(0,0);
        }

        //public void accumulate(ACC accumulator, [user defined inputs])
        public void accumulate(MyAcc accumulator, Integer vc){

            //如果vc > firstVc ,把之前的firstVc 调整为 secondVc，把vc调整为 firstVc
            if (vc > accumulator.firstVc){
                accumulator.setSecondVc(accumulator.firstVc);
                accumulator.setFirstVc(vc);
            }else if (vc > accumulator.secondVc){
                accumulator.setSecondVc(vc);
            }

        }

        //输出
        public void emitValue(MyAcc acc, Collector<Row> out){
            out.collect(Row.of(1,acc.firstVc));
            //out 调用一次，就是输出一行
            if (acc.secondVc > 0){
                //说明当前这种类型的传感器，至少有前2名
                out.collect(Row.of(2,acc.secondVc));
            }
        }


    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer firstVc;
        private Integer secondVc;
    }
}
